﻿/************************************************************************

   Extended Silverlight Toolkit

   Copyright (C) 2010-2012 Xceed Software Inc.

   This program is provided to you under the terms of the Microsoft Public
   License (Ms-PL) as published at http://extendedsilverlight.codeplex.com/license

   Please purchase a commercial version of this toolkit if you are using
   it in a commercial product. The commercial versions support the project,
   and also include more features for each control, priority support,
   source code, updates, removed watermark, and a proprietary license.

   Visit http://xceed.com and follow @datagrid on Twitter.

  **********************************************************************/

using System;
using System.Collections;
using System.Collections.Generic;
using System.ComponentModel;
using System.Diagnostics;
using System.Linq;
using System.Windows.Threading;

namespace Xceed.Silverlight.Data
{
  internal abstract class ReactiveEnumerator<T> : SyncEnumerator<T>, IPendingRequests
  {
    #region Constructor

    protected ReactiveEnumerator(
      IAsyncEnumerator<T> childEnumerator,
      DataItemFactory<T> dataItemFactory,
      INotifyDataSourceChanged dataSourceChanged,
      ThrottlingHandler throttlingHandler )
    {
      if( childEnumerator == null )
        throw new ArgumentNullException( "childEnumerator" );

      if( dataItemFactory == null )
        throw new ArgumentNullException( "dataItemFactory" );

      m_childEnumerator = childEnumerator;
      m_childEnumerator.DataChanged += new EventHandler<DataChangedEventArgs>( this.OnChildDataChanged );

      this.DataItemFactory = dataItemFactory;

      m_dataSourceChanged = dataSourceChanged;
      if( m_dataSourceChanged != null )
      {
        m_dataSourceChanged.DataSourceChanged += new EventHandler<DataSourceChangedEventArgs>( this.OnDataSourceChanged );
      }

      m_throttlingHandler = throttlingHandler;

      m_processOperationTimer = new DispatcherTimer();
      m_processOperationTimer.Tick += new EventHandler( this.ProcessOperations );
    }

    #endregion

    #region DataItemFactory Property

    internal DataItemFactory<T> DataItemFactory
    {
      get;
      private set;
    }

    #endregion

    #region QueueCount Private Property

    private int QueueCount
    {
      get
      {
        return m_operationsQueue.Count + m_statQueries.Count;
      }
    }

    #endregion

    #region IsQueueEmpty Private Property

    private bool IsQueueEmpty
    {
      get
      {
        return ( this.QueueCount == 0 );
      }
    }

    #endregion

    #region OperationCompleted Event

    internal event EventHandler<OperationCompletedEventArgs<T>> OperationCompleted;

    private void RaiseOperationCompleted( 
      DataSourceContext context,
      JumpMoveAndFetchResult<T> operationResult, 
      IList<Tuple<T, T>> replacements, 
      IList<T> deletions )
    {
      if( this.OperationCompleted != null )
      {
        this.OperationCompleted( this, new OperationCompletedEventArgs<T>( context, operationResult, replacements, deletions ) );
      }
    }

    private void RaiseOperationCompleted( 
      StatParameters statParameters, 
      StatResults dummyResults, 
      StatResults realResults )
    {
      if( this.OperationCompleted != null )
      {
        this.OperationCompleted( this, new OperationCompletedEventArgs<T>( statParameters, dummyResults, realResults ) );
      }
    }

    #endregion

    #region Unsupported Count methods

    public override CountResult Count( CountParameters parameters )
    {
      throw new NotSupportedException();
    }

    public override GroupCountResult GroupCount( CountParameters parameters )
    {
      throw new NotSupportedException();
    }

    #endregion

    #region GetStatValue Method

    public override StatResults GetStatValues( StatParameters parameters )
    {
      StatOperationInfo statOperationInfo = new StatOperationInfo( parameters, this.GetCurrentContext() );
      StatResults returnedValue;

      if( this.EnqueueOperation( statOperationInfo ) )
      {
        returnedValue = statOperationInfo.RealResults;
      }
      else
      {
        returnedValue = this.DataItemFactory.CreateDummyStatResults( parameters );
        statOperationInfo.DummyResults = returnedValue;
      }

      return returnedValue;
    }

    #endregion

    #region MoveAndFetch Method

    public override bool JumpMoveAndFetch(
      JumpMoveAndFetchParameters parameters,
      out Nullable<long> actualMoveCount,
      out ICollection<T> fetchedItems,
      out SourcePosition endPosition,
      out RelativeOffset endPositionOffset,
      out EstimateInt64 totalCount )
    {
      this.CheckMoveAndFetchParameters( parameters.ExpectedMoveCount, parameters.ExpectedFetchCount );

      // Make sure the opertion doesn't stay out of bound.
      if( this.HandleOutOfBoundOperation( parameters, out actualMoveCount, out fetchedItems, out endPosition, out endPositionOffset, out totalCount ) )
        return false;

      ReactiveOperationInfo<T> operationInfo = new ReactiveOperationInfo<T>(
                                                 parameters.SourcePosition,
                                                 parameters.ExpectedMoveCount,
                                                 parameters.ExpectedFetchCount,
                                                 parameters.ApproximateMove,
                                                 this.GetCurrentContext() );

      // The operation has been done synchronously.
      if( this.EnqueueOperation( operationInfo ) )
      {
        actualMoveCount = operationInfo.ActualMoveCount;
        fetchedItems = operationInfo.FetchedItems;
        endPosition = operationInfo.EndPosition;
        endPositionOffset = operationInfo.EndPositionOffset;
        totalCount = operationInfo.TotalCount;
      }
      // The operation has been enqueued and will be executed in another time.
      else
      {
        T dummyTemplate = this.CreateDummyTemplate( operationInfo.Context, parameters.SourcePosition );

        actualMoveCount = operationInfo.ActualMoveCount.GetValueOrDefault( parameters.ExpectedMoveCount );
        fetchedItems = this.CreateDummyItems( dummyTemplate, parameters.ExpectedFetchCount );

        operationInfo.FetchedItems = fetchedItems;
        Debug.Assert( fetchedItems != null );

        if( fetchedItems.Count > 0 )
        {
          endPosition = new DummyPosition<T>( fetchedItems.Last() );
        }
        else
        {
          Debug.Assert( parameters.ExpectedMoveCount != 0 );

          endPosition = new DummyPosition<T>(
                          parameters.SourcePosition,
                          parameters.ExpectedMoveCount,
                          parameters.ApproximateMove );
        }

        operationInfo.EndPosition = endPosition;
      }

      return operationInfo.Result;
    }

    private bool HandleOutOfBoundOperation(
      JumpMoveAndFetchParameters parameters,
      out Nullable<long> actualMoveCount,
      out ICollection<T> fetchedItems,
      out SourcePosition endPosition,
      out RelativeOffset endPositionOffset,
      out EstimateInt64 totalCount )
    {
      SourcePosition sourcePosition = parameters.SourcePosition;
      long moveCount = parameters.ExpectedMoveCount;

      // The operation will fail since we'll stay out of bound.
      if( ( moveCount <= 0 && sourcePosition.IsBeforeFirst )
        || ( moveCount >= 0 && sourcePosition.IsAfterLast ) )
      {
        actualMoveCount = 0;
        fetchedItems = null;
        endPosition = sourcePosition;
        totalCount = null;

        ReferencePoint referencePoint = ( sourcePosition.IsBeforeFirst ) ? ReferencePoint.BeforeFirst : ReferencePoint.AfterLast;
        endPositionOffset = new RelativeOffset( 0, referencePoint );

#if DEBUG
        //This is needed for the unit tests.
        DataEnumeratorResultEventArgs eventArgs = new DataEnumeratorResultEventArgs(
          new JumpMoveAndFetchResult<T>(
            parameters.SourcePosition,
            parameters.ApproximateMove,
            parameters.ExpectedMoveCount,
            parameters.ExpectedFetchCount,
            actualMoveCount,
            fetchedItems,
            false,
            endPosition,
            endPositionOffset,
            totalCount ) );

        this.OnMoveAndFetchCompleted( eventArgs );
#endif

        return true;
      }
      else
      {
        actualMoveCount = null;
        fetchedItems = null;
        endPosition = null;
        endPositionOffset = null;
        totalCount = null;

        return false;
      }
    }

    private bool ProcessOperation( StatOperationInfo statOperationInfo )
    {
      IAsyncResult asyncResult;

      try
      {
        m_isProcessingOperation = true;

        asyncResult = m_childEnumerator.BeginGetStatValues(
                        statOperationInfo.Parameters,
                        this.GetStatValueCallback,
                        statOperationInfo );
      }
      catch( Exception )
      {
        m_isProcessingOperation = false;
        throw;
      }

      if( ( asyncResult.IsCompleted ) && ( asyncResult.CompletedSynchronously ) )
      {
        this.GetStatValueCompletion( asyncResult );
      }

      return asyncResult.IsCompleted;
    }

    private bool ProcessOperation( ReactiveOperationInfo<T> operationInfo )
    {
      if( operationInfo == null )
        throw new ArgumentNullException( "operationInfo" );

      DummyPosition<T> dummySourcePosition = operationInfo.SourcePosition as DummyPosition<T>;

      SourcePosition sourcePosition;
      T item;
      long offset;
      bool approximateOffset;
      Func<T, SourcePosition> positionResolver;

      bool isBasedOnItem = DataUtils.ResolvePosition<T>(
        operationInfo.SourcePosition,
        true,
        out sourcePosition,
        out item,
        out offset,
        out approximateOffset,
        out positionResolver );

      if( isBasedOnItem )
      {
        if( this.DataItemFactory.IsDummy( item ) )
          throw new DataInternalException();// Unresolved dummy item is not used.

        if( sourcePosition == null )
        {
          Debug.Assert( positionResolver == null, "This should have been handled in UpdateQueuedOperations" );
          sourcePosition = new ItemPosition( item );
        }
      }

      operationInfo.SourcePosition = this.GetNewSourcePosition( operationInfo.Context, sourcePosition );
      operationInfo.ExpectedMoveCount += offset;
      operationInfo.ApproximateMove = ( operationInfo.ApproximateMove || approximateOffset );

      Debug.Assert( operationInfo.ExpectedFetchCount != 0 || operationInfo.ExpectedMoveCount != 0 );

      IAsyncResult asyncResult;

      try
      {
        m_isProcessingOperation = true;

        asyncResult = m_childEnumerator.BeginJumpMoveAndFetch(
                        operationInfo.GetParameters(),
                        this.MoveAndFetchCallback,
                        operationInfo );
      }
      catch( Exception )
      {
        m_isProcessingOperation = false;
        throw;
      }

      if( ( asyncResult.IsCompleted ) && ( asyncResult.CompletedSynchronously ) )
      {
        this.MoveAndFetchCompletion( asyncResult );
      }

      return asyncResult.IsCompleted;
    }

    private void GetStatValueCallback( IAsyncResult asyncResult )
    {
      if( asyncResult.CompletedSynchronously )
        return;

      this.GetStatValueCompletion( asyncResult );
      this.DequeueOperation( true );
    }

    private void MoveAndFetchCallback( IAsyncResult asyncResult )
    {
      if( ( asyncResult.IsCompleted ) && ( asyncResult.CompletedSynchronously ) )
        return;

      this.MoveAndFetchCompletion( asyncResult );
      this.DequeueOperation( false );
    }

    private void GetStatValueCompletion( IAsyncResult asyncResult )
    {
      StatOperationInfo statOperationInfo = ( StatOperationInfo )asyncResult.AsyncState;

      try
      {
        statOperationInfo.RealResults = m_childEnumerator.EndGetStatValues( asyncResult );

        if( m_disposed )
          return;

        if( statOperationInfo.DummyResults != null )
        {
          this.RaiseOperationCompleted( statOperationInfo.InitialParameters, statOperationInfo.DummyResults, statOperationInfo.RealResults );
        }
      }
      finally
      {
        m_isProcessingOperation = false;
      }
    }

    private void MoveAndFetchCompletion( IAsyncResult asyncResult )
    {
      if( m_disposed )
        return;

      JumpMoveAndFetchResult<T> result = null;
      ReactiveOperationInfo<T> operationInfo = ( ReactiveOperationInfo<T> )asyncResult.AsyncState;

      List<Tuple<T, T>> replacements = new List<Tuple<T, T>>();
      List<T> deletions = new List<T>();

      try
      {
        Debug.Assert( object.ReferenceEquals( m_operationsQueue[ 0 ], operationInfo ) );

        result = m_childEnumerator.EndJumpMoveAndFetch( ( JumpMoveAndFetchAsyncResult )asyncResult );

        bool synchronousOperation = ( operationInfo.FetchedItems == null );
        if( synchronousOperation )
        {
          Debug.Assert( asyncResult.CompletedSynchronously );

          // Performed synchronously. Use the extracted fetched items as-is.
          operationInfo.FetchedItems = result.FetchedItems;
        }
        else
        {
          List<T> dummyItems = operationInfo.FetchedItems.ToList();
          List<T> realItems = ( result.FetchedItems != null )
                                ? result.FetchedItems.ToList()
                                : new List<T>( 0 );

          Debug.Assert( dummyItems.Count >= realItems.Count, @"How can we have more items than requested?" );

          DummyPosition<T> dummyEndPosition = ( DummyPosition<T> )operationInfo.EndPosition;
          dummyEndPosition.RealPosition = result.EndPositionObject;

          for( int i = 0; i < realItems.Count; i++ )
          {
            replacements.Add( new Tuple<T, T>( dummyItems[ i ], realItems[ i ] ) );
          }

          deletions.AddRange( dummyItems.GetRange( realItems.Count, dummyItems.Count - realItems.Count ) );

          this.UpdateQueuedOperations(
            dummyItems,
            realItems,
            dummyEndPosition,
            result.EndPositionObject,
            deletions );
        }

        operationInfo.EndPosition = result.EndPositionObject;
        operationInfo.ActualMoveCount = result.ActualMoveCount;
        operationInfo.Result = result.Result;
        operationInfo.EndPositionOffset = result.EndPositionOffset;
        operationInfo.TotalCount = result.TotalCount;

        if( !synchronousOperation )
        {
          this.RaiseOperationCompleted( operationInfo.Context, result, replacements, deletions );
        }

#if DEBUG
        if( this.MoveAndFetchCompleted != null )
        {
          this.MoveAndFetchCompleted( this, new DataEnumeratorResultEventArgs( result ) );
        }
#endif
      }
      finally
      {
        m_isProcessingOperation = false;
      }
    }

#if DEBUG
    [EditorBrowsable( EditorBrowsableState.Never )]
    public event EventHandler<DataEnumeratorResultEventArgs> MoveAndFetchCompleted;

    [EditorBrowsable( EditorBrowsableState.Never )]
    private void OnMoveAndFetchCompleted( DataEnumeratorResultEventArgs e )
    {
      if( this.MoveAndFetchCompleted != null )
      {
        this.MoveAndFetchCompleted( this, e );
      }
    }
#endif

    #endregion MoveAndFetch Method

    protected virtual T CreateDummyItem( T templateItem )
    {
      return this.DataItemFactory.CreateInstance( templateItem );
    }

    protected virtual T CreateDummyTemplate( DataSourceContext context, SourcePosition sourcePosition )
    {
      T item;
      SourcePosition position;
      long offset;
      bool approximateOffset;
      Func<T, SourcePosition> positionResolver;

      DataUtils.ResolvePosition<T>(
        sourcePosition,
        true,
        out position,
        out item,
        out offset,
        out approximateOffset,
        out positionResolver );

      return ( positionResolver == null ) ? item : default( T );
    }

    protected virtual bool DummyDependOn( T doesThisDummy, T dependOnThisOne )
    {
      return object.Equals( doesThisDummy, dependOnThisOne );
    }

    protected virtual bool ShouldPerformFetch( T item )
    {
      return true;
    }

    protected virtual SourcePosition GetNewSourcePosition( DataSourceContext context, SourcePosition position )
    {
      return position;
    }

    protected override void Dispose( bool disposing )
    {
      m_disposed = true;

      // Do not process any more queries.
      m_isProcessingOperation = false;

      // Empty the query queues.
      m_operationsQueue.Clear();
      m_statQueries.Clear();

      // Dispose the child enumerator.
      m_childEnumerator.DataChanged -= new EventHandler<DataChangedEventArgs>( this.OnChildDataChanged );
      m_childEnumerator.Dispose();

      if( m_dataSourceChanged != null )
      {
        m_dataSourceChanged.DataSourceChanged -= new EventHandler<DataSourceChangedEventArgs>( this.OnDataSourceChanged );
      }

      // Remove all listeners.
      this.OperationCompleted = null;
      this.HasPendingRequestsChanged = null;
#if DEBUG
      this.MoveAndFetchCompleted = null;
#endif

      base.Dispose( disposing );
    }

    protected abstract DataSourceContext GetCurrentContext();

    private ICollection<T> CreateDummyItems( T templateItem, int expectedFetchCount )
    {
      int itemCount = Math.Abs( expectedFetchCount );
      List<T> items = new List<T>( itemCount );

      for( int i = 0; i < itemCount; i++ )
      {
        T item = this.CreateDummyItem( templateItem );
        if( i == 0 )
        {
          templateItem = item;
        }

        items.Add( item );
      }

      return items;
    }

    private void UpdateQueuedOperations( List<T> dummyItems, List<T> realItems,
      DummyPosition<T> dummyPosition, SourcePosition realPosition, List<T> deletions )
    {
      if( dummyItems == null )
        throw new ArgumentNullException( "dummyItems" );

      if( deletions == null )
        throw new ArgumentNullException( "deletions" );

      if( realItems == null )
        realItems = new List<T>( 0 );

      List<ReactiveOperationInfo<T>> newQueue = new List<ReactiveOperationInfo<T>>();

      for( int i = 0; i < m_operationsQueue.Count; i++ )
      {
        bool isInvalid = false;
        ReactiveOperationInfo<T> jumpOperation = m_operationsQueue[ i ];

        if( this.AreEquivalent( jumpOperation.SourcePosition, dummyPosition, false ) )
        {
          jumpOperation.SourcePosition = realPosition;
        }
        else
        {
          T basedItem;
          long offset;
          SourcePosition sourcePosition;
          bool approximateOffset;
          Func<T, SourcePosition> positionResolver;

          bool isBasedOnItem = DataUtils.ResolvePosition<T>(
                                 jumpOperation.SourcePosition,
                                 true,
                                 out sourcePosition,
                                 out basedItem,
                                 out offset,
                                 out approximateOffset,
                                 out positionResolver );

          if( isBasedOnItem )
          {
            //If this opertion SourcePosition is based on an invalid item, just invalidate and remove
            //all fetched items in this operation.
            if( deletions.Any( ( deleted ) => this.DummyDependOn( basedItem, deleted ) ) )
            {
              isInvalid = true;

              if( jumpOperation.FetchedItems != null )
              {
                deletions.AddRange( jumpOperation.FetchedItems );
              }
            }
            else
            {
              int dummyIndex = -1;
              for( int j = 0; j < dummyItems.Count; j++ )
              {
                if( this.DummyDependOn( basedItem, dummyItems[ j ] ) )
                {
                  dummyIndex = j;
                  break;
                }
              }
              //int dummyIndex = dummyItems.IndexOf( basedItem );

              if( dummyIndex > -1 && dummyIndex < realItems.Count )
              {
                if( positionResolver != null )
                {
                  sourcePosition = positionResolver( realItems[ dummyIndex ] );
                  jumpOperation.SourcePosition = new DummyPosition<T>(
                                                       sourcePosition,
                                                       offset,
                                                       approximateOffset );
                }
                else
                {
                  if( offset == 0 )
                  {
                    jumpOperation.SourcePosition = new ItemPosition( realItems[ dummyIndex ] );
                  }
                  else
                  {
                    jumpOperation.SourcePosition = new DummyPosition<T>(
                                                         realItems[ dummyIndex ],
                                                         offset,
                                                         approximateOffset );
                  }
                }
              }
            }
          }
        }

        if( !isInvalid )
        {
          newQueue.Add( jumpOperation );
        }
      }

      m_operationsQueue = newQueue;
    }

    private bool ShouldPerformFetches( ICollection<T> items )
    {
      if( items == null || items.Count <= 0 )
        return false;

      foreach( T item in items )
      {
        if( this.ShouldPerformFetch( item ) )
          return true;
      }

      return false;
    }

    private void ProcessOperations( object sender, EventArgs e )
    {
      DispatcherTimer timer = ( DispatcherTimer )sender;
      timer.Stop();

      if( !m_disposed )
      {
        if( m_isProcessingOperation )
          return;

        this.ProcessOperations();

        if( !this.IsQueueEmpty )
        {
          this.RaiseHasPendingRequestsChanged();
        }
      }
      else
      {
        timer.Tick -= new EventHandler( this.ProcessOperations );
      }
    }

    private bool ProcessOperations()
    {
      Debug.Assert( !m_processOperationTimer.IsEnabled );

      bool synchronousOperation = false;

      while( true )
      {
        this.OptimizeOperationsQueue();

        // Leave the loop if all operations have been processed
        // or have been removed because of optimizations.
        if( this.IsQueueEmpty )
          break;

        if( m_operationsQueue.Count > 0 )
        {
          synchronousOperation = this.ProcessOperation( m_operationsQueue[ 0 ] );

          // Only one asynchronous operation may be done.
          if( !synchronousOperation )
            break;

          m_operationsQueue.RemoveAt( 0 );
        }
        else
        {
          synchronousOperation = this.ProcessOperation( m_statQueries[ 0 ] );

          // Only one asynchronous operation may be done.
          if( !synchronousOperation )
            break;

          m_statQueries.RemoveAt( 0 );
        }
      }

      return synchronousOperation;
    }

    private bool EnqueueOperation( StatOperationInfo statOperationInfo )
    {
      if( m_disposed )
        return false;

      m_statQueries.Add( statOperationInfo );

      return this.HandleEnqueueOperation();
    }

    private bool EnqueueOperation( ReactiveOperationInfo<T> operationInfo )
    {
      if( m_disposed )
        return false;

      m_operationsQueue.Add( operationInfo );

      return this.HandleEnqueueOperation();
    }

    private bool HandleEnqueueOperation()
    {
      if( m_isProcessingOperation || m_processOperationTimer.IsEnabled )
        return false;

      TimeSpan waitDelay = ( m_throttlingHandler != null )
                             ? m_throttlingHandler.RequestExecutionDelay
                             : TimeSpan.Zero;

      bool synchronousOperation = false;

      if( waitDelay.TotalMilliseconds < 1d )
      {
        synchronousOperation = this.ProcessOperations();

        if( this.QueueCount == 1 )
        {
          this.RaiseHasPendingRequestsChanged();
        }
      }
      else
      {
        m_processOperationTimer.Interval = waitDelay;
        m_processOperationTimer.Start();
      }

      return synchronousOperation;
    }

    private void DequeueOperation( bool statOperation )
    {
      if( m_disposed )
        return;

      if( statOperation )
      {
        m_statQueries.RemoveAt( 0 );
      }
      else
      {
        m_operationsQueue.RemoveAt( 0 );
      }

      this.ProcessOperations();

      if( this.IsQueueEmpty )
      {
        this.RaiseHasPendingRequestsChanged();
      }
    }

    private int GetQueueCount()
    {
      return ( m_operationsQueue.Count + m_statQueries.Count );
    }

    private void OptimizeOperationsQueue()
    {
      if( m_operationsQueue.Count <= 1 )
        return;

      // Remove useless operations and merged them in one big move.
      // It is very useful when scrolling lots of items at a time.
      this.OptimizeMoveOperations();

      // Merge many consecutive move, fetch, move, fetch into fewer move/fetch.
      this.OptimizeFetchOperations();
    }

    protected bool AreEquivalent(
      SourcePosition first,
      SourcePosition second,
      bool considerRealPosition )
    {
      if( object.Equals( first, second ) )
        return true;

      if( first == null || second == null )
        return false;

      long firstRealPositionOffset = 0;
      long secondRealPositionOffset = 0;

      bool firstApproximateOffset;
      bool secondApproximateOffset;

      T firstItem;
      T secondItem;

      SourcePosition firstRealPosition = first;
      SourcePosition secondRealPosition = second;

      Func<T, SourcePosition> firstPositionResolver;
      Func<T, SourcePosition> secondPositionResolver;

      bool firstBasedOnItem = DataUtils.ResolvePosition<T>(
                                first,
                                considerRealPosition,
                                out firstRealPosition,
                                out firstItem,
                                out firstRealPositionOffset,
                                out firstApproximateOffset,
                                out firstPositionResolver );
      bool secondBasedOnItem = DataUtils.ResolvePosition<T>(
                                 second,
                                 considerRealPosition,
                                 out secondRealPosition,
                                 out secondItem,
                                 out secondRealPositionOffset,
                                 out secondApproximateOffset,
                                 out secondPositionResolver );

      if( firstBasedOnItem != secondBasedOnItem )
        return false;

      if( firstRealPositionOffset != secondRealPositionOffset )
        return false;

      if( ( firstRealPositionOffset != 0 ) && firstApproximateOffset != secondApproximateOffset )
        return false;

      if( !object.Equals( firstPositionResolver, secondPositionResolver ) )
        return false;

      return ( firstBasedOnItem )
        ? object.Equals( firstItem, secondItem )
        : object.Equals( firstRealPosition, secondRealPosition );
    }

    private bool IsNeededOperation( ReactiveOperationInfo<T> operation )
    {
      if( operation == null )
        throw new ArgumentNullException( "operation" );

      return ( operation.ExpectedFetchCount != 0 ) && this.ShouldPerformFetches( operation.FetchedItems );
    }

    private void OptimizeMoveOperations()
    {
      //Need at least 2 requests to optimize.
      if( m_operationsQueue.Count < 2 )
        return;

      //This class contain the list needed operations that depends on dummy items to be executed.
      DependenceInformation<T> dependenceInformation = new DependenceInformation<T>();

      //Contains a list of all the needed operations that must not be flushed by this call.
      List<ReactiveOperationInfo<T>> neededOperations = new List<ReactiveOperationInfo<T>>();

      T dummy;
      long offset;
      bool approximateOffset;

      for( int i = m_operationsQueue.Count - 1; i >= 0; i-- )
      {
        ReactiveOperationInfo<T> currentOperation = m_operationsQueue[ i ];
        List<T> currentOperationDummies = currentOperation.FetchedItems.ToList();

        //The last operation is always needed, it contains the last request of the user and possibly a dummy end position.
        bool isNeededOperation =
          ( i == m_operationsQueue.Count - 1 ) ||
          this.IsNeededOperation( currentOperation );

        SourcePosition notUsed;
        Func<T, SourcePosition> positionResolver;
        bool dependOnItem = DataUtils.ResolvePosition<T>(
                                            currentOperation.SourcePosition,
                                            true,
                                            out notUsed,
                                            out dummy,
                                            out offset,
                                            out approximateOffset,
                                            out positionResolver );

        bool operationDependOnDummyItem = dependOnItem && ( this.DataItemFactory.IsDummy( dummy ) );
        bool hardDependence = ( positionResolver != null );

        for( int j = 0; ( !isNeededOperation ) && ( j < currentOperationDummies.Count ); j++ )
        {
          T currentDummy = currentOperationDummies[ j ];
          if( dependenceInformation.HaveHardDependenceOn( currentDummy, this.DummyDependOn ) )
          {
            isNeededOperation = true;
          }
        }

        if( isNeededOperation )
        {
          neededOperations.Add( currentOperation );
          if( operationDependOnDummyItem )
          {
            dependenceInformation.AddDependent( dummy, i, offset, approximateOffset, hardDependence );
          }
        }

        for( int j = 0; j < currentOperationDummies.Count; j++ )
        {
          T currentDummy = currentOperationDummies[ j ];
          int currentDummyOffset = j * DataUtils.GetUnit( currentOperation.ExpectedFetchCount );
          IList<DependentOperationInfo<T>> dependentOperations = new List<DependentOperationInfo<T>>();

          dependenceInformation.RemoveDependentOperations( currentDummy, this.DummyDependOn, dependentOperations, null );

          foreach( DependentOperationInfo<T> operationInfo in dependentOperations )
          {
            if( isNeededOperation )
            {
              ReactiveOperationInfo<T> dependentOperation = m_operationsQueue[ operationInfo.Index ];

              T dependentDummy;
              long dependentOffset;
              bool dependentApprox;
              SourcePosition dependentSourcePosition;
              Func<T, SourcePosition> dependentPositionResolver;
              if( !DataUtils.ResolvePosition<T>( dependentOperation.SourcePosition, true, out dependentSourcePosition, out dependentDummy, out dependentOffset, out dependentApprox, out dependentPositionResolver ) )
              {
                Debug.Assert( false, @"Not supposed to depend on anything if source position don't depend on dummy" );
              }

              Debug.Assert( !object.Equals( currentDummy, dependentDummy )
                || ( ( operationInfo.Offset == dependentOffset ) && ( operationInfo.ApproximateOffset == dependentApprox ) ),
                "Dependence on a dummy is not consistent with initial logged values" );

              if( operationInfo.Offset != dependentOffset
                  || operationInfo.ApproximateOffset != dependentApprox
                  || !object.Equals( currentDummy, dependentDummy ) )
              {
                dependentOperation.SourcePosition = new DummyPosition<T>( currentDummy, operationInfo.Offset, operationInfo.ApproximateOffset );
              }
            }
            else
            {
              bool newApproximateOffset = false;

              if( operationInfo.Offset != 0 )
              {
                newApproximateOffset = operationInfo.ApproximateOffset;
              }

              if( currentOperation.ExpectedMoveCount != 0 )
              {
                newApproximateOffset = newApproximateOffset || currentOperation.ApproximateMove;
              }

              if( operationDependOnDummyItem )
              {
                newApproximateOffset = newApproximateOffset || approximateOffset;
                dependenceInformation.AddDependent(
                  dummy, operationInfo.Index,
                  operationInfo.Offset + currentDummyOffset + offset + currentOperation.ExpectedMoveCount,
                  newApproximateOffset, false );
              }
              else
              {
                ReactiveOperationInfo<T> dependentOperation = m_operationsQueue[ operationInfo.Index ];
                dependentOperation.SourcePosition = currentOperation.SourcePosition;
                dependentOperation.ExpectedMoveCount += ( operationInfo.Offset + currentDummyOffset + currentOperation.ExpectedMoveCount );
                dependentOperation.ApproximateMove |= newApproximateOffset;
              }
            }
          }
        }
      }

      //Clear the list of all the un needed operations
      neededOperations.Reverse();

      m_operationsQueue = neededOperations;

      if( !dependenceInformation.IsEmpty )
        throw new DataInternalException(); //parentdummy item will not be fetched. 
    }

    private void OptimizeFetchOperations()
    {
      bool firstFetchOccurred = false;
      int fetchedItemDirection = 1;
      bool preFetchMoveApprox = false;
      long startIndex = 0;
      long endIndex = 0;
      long currentIndex = 0;
      List<T> fetchedItems = new List<T>();
      bool isOperationOptimizable = true;
      SourcePosition initialSourcePosition = null;
      SourcePosition moveAndFetchEndPosition = null;
      ReactiveOperationInfo<T> previousOperationInfo = null;
      DataSourceContext context = null;

      while( ( isOperationOptimizable ) && ( m_operationsQueue.Count > 0 ) )
      {
        isOperationOptimizable = false;
        ReactiveOperationInfo<T> operationInfo = m_operationsQueue[ 0 ];

        if( previousOperationInfo != null )
        {
          // We cannot optimize two consecutive operations if they aren't related.
          if( !this.AreEquivalent( previousOperationInfo.EndPosition, operationInfo.SourcePosition, false )
            || !object.Equals( previousOperationInfo.Context, operationInfo.Context ) )
            break;
        }

        if( initialSourcePosition == null )
        {
          initialSourcePosition = operationInfo.SourcePosition;
        }

        if( context == null )
        {
          context = operationInfo.Context;
        }

        long moveCount = operationInfo.ExpectedMoveCount;
        int fetchCount = operationInfo.ExpectedFetchCount;
        int fetchDisplacement = 0;

        Debug.Assert( Math.Abs( fetchCount ) == operationInfo.FetchedItems.Count );

        // Moves are optimizable if we aren't optimizing the fetch part and
        // we aren't creating any gap.
        isOperationOptimizable = ( moveCount == 0 )
                              || ( !firstFetchOccurred )
                              || ( !operationInfo.ApproximateMove );

        if( ( isOperationOptimizable ) && ( operationInfo.ExpectedFetchCount != 0 ) )
        {
          isOperationOptimizable = false;

          int currentFetchDirection = DataUtils.GetUnit( operationInfo.ExpectedFetchCount );

          fetchDisplacement = operationInfo.ExpectedFetchCount - currentFetchDirection;

          // The fetching part will be the starting point of the batch.
          if( !firstFetchOccurred )
          {
            firstFetchOccurred = true;
            fetchedItemDirection = currentFetchDirection;
            fetchedItems.AddRange( operationInfo.FetchedItems );

            startIndex = currentIndex + moveCount;
            endIndex = startIndex + fetchDisplacement;
            isOperationOptimizable = true;
          }
          // Append the fetching items to the current batch if there is no gap.
          else
          {
            long fetchFirstItemIndex = currentIndex + moveCount;
            long fetchLastItemIndex = fetchFirstItemIndex + fetchDisplacement;

            if( currentFetchDirection == fetchedItemDirection )
            {
              //The current fetch is in the same direction of the
              //current stacked items

              if( fetchFirstItemIndex == ( endIndex + fetchedItemDirection ) )
              {
                //Current fetch is continuous with the stacked items end.
                fetchedItems.AddRange( operationInfo.FetchedItems );
                endIndex = fetchLastItemIndex;
                isOperationOptimizable = true;
              }
              else if( fetchLastItemIndex == ( startIndex - fetchedItemDirection ) )
              {
                //Current fetch is continuous with the stacked items start.
                fetchedItems.InsertRange( 0, operationInfo.FetchedItems );
                startIndex = fetchFirstItemIndex;
                isOperationOptimizable = true;
              }
            }
            else
            {
              //The current fetch is in the opposite direction of the
              //current stacked items
              if( fetchFirstItemIndex == ( startIndex - fetchedItemDirection ) )
              {
                //Current fetch is continuous with the stacked items start.
                fetchedItems.InsertRange( 0, operationInfo.FetchedItems.Reverse() );
                startIndex = fetchLastItemIndex;
                isOperationOptimizable = true;
              }
              else if( fetchLastItemIndex == ( endIndex + fetchedItemDirection ) )
              {
                //Current fetch is continuous with the stacked items end.
                fetchedItems.AddRange( operationInfo.FetchedItems.Reverse() );
                endIndex = fetchFirstItemIndex;
                isOperationOptimizable = true;
              }
            }
          }
        }

        if( isOperationOptimizable )
        {
          // Remove the operation since there might be a simplification.
          if( moveCount != 0 )
          {
            currentIndex += moveCount;
            preFetchMoveApprox = preFetchMoveApprox || operationInfo.ApproximateMove;
          }
          currentIndex += fetchDisplacement;
          moveAndFetchEndPosition = operationInfo.EndPosition;

          m_operationsQueue.RemoveAt( 0 );
        }

        previousOperationInfo = operationInfo;
      }

      // Note that this algorithm will sometimes end up removing an operation only 
      // to reinsert it at the same position when no optimization is needed. This 
      // was intentionally left like this. It could be corrected if we ever measure that
      // the solution would be more performant than removing and adding an item in a list.
      if( firstFetchOccurred )
      {
        Debug.Assert( ( endIndex == startIndex ) || ( DataUtils.GetUnit( endIndex - startIndex ) == fetchedItemDirection ), "Something is wrong with the algorithm" );
        Debug.Assert( true, "We should check that moveAndFetchEndPosition represent the same place than currentIndex" );

        // Queue the fetch or move/fetch operation that will occur first.
        ReactiveOperationInfo<T> operationInfo = new ReactiveOperationInfo<T>(
                                                   initialSourcePosition,
                                                   startIndex,
                                                   Convert.ToInt32( endIndex - startIndex ) + fetchedItemDirection,
                                                   preFetchMoveApprox,
                                                   context );
        operationInfo.FetchedItems = fetchedItems;
        operationInfo.EndPosition = new DummyPosition<T>( fetchedItems.Last() );
        m_operationsQueue.Insert( 0, operationInfo );

        long postFetchMove = currentIndex - endIndex;

        if( postFetchMove != 0 )
        {
          ReactiveOperationInfo<T> moveOperationInfo = new ReactiveOperationInfo<T>(
                                                         operationInfo.EndPosition,
                                                         postFetchMove,
                                                         0,
                                                         true,
                                                         context );
          moveOperationInfo.FetchedItems = new List<T>();
          moveOperationInfo.EndPosition = moveAndFetchEndPosition;
          m_operationsQueue.Insert( 1, moveOperationInfo );
        }
      }
      else if( currentIndex != 0 )
      {
        Debug.Assert( true, "We should check that moveAndFetchEndPosition represent the same place than currentIndex" );

        ReactiveOperationInfo<T> moveOperationInfo = new ReactiveOperationInfo<T>(
                                                       initialSourcePosition,
                                                       currentIndex,
                                                       0,
                                                       preFetchMoveApprox,
                                                       context );
        moveOperationInfo.FetchedItems = new List<T>();

        // This is not an approximate move since we do not optimize
        // approximate move position.
        moveOperationInfo.EndPosition = moveAndFetchEndPosition;
        m_operationsQueue.Insert( 0, moveOperationInfo );
      }
    }

    private void OnChildDataChanged( object sender, DataChangedEventArgs e )
    {
      this.RaiseDataChanged( e );
    }

    private void OnDataSourceChanged( object sender, DataSourceChangedEventArgs e )
    {
      if( !m_disposed )
      {
        foreach( DataSourceChangedEventArgs dataSourceChange in m_childEnumerator.TransformDataSourceChangedEventArgs( this.GetCurrentContext(), e ) )
        {
          this.OnDataChanged( dataSourceChange );
        }
      }
    }

    private void OnDataChanged( DataSourceChangedEventArgs e )
    {
      if( ( m_disposed ) || ( e == null ) )
        return;

      if( e.Action == DataSourceChangedAction.Add )
      {
        foreach( T item in e.NewItems.Cast<T>() )
        {
          this.RaiseDataAdd( item );
        }
      }
      else if( e.Action == DataSourceChangedAction.Remove )
      {
        foreach( T item in e.OldItems.Cast<T>() )
        {
          this.RaiseDataRemoved( item, null, null );
        }
      }
      else if( e.Action == DataSourceChangedAction.Replace )
      {
        if( e.NewItems.Count != e.OldItems.Count )
          throw new ArgumentException( "The new item count is different than the old item count.", "e" );

        int count = e.NewItems.Count;
        for( int i = 0; i < count; i++ )
        {
          T oldItem = ( T )e.OldItems[ i ];
          T newItem = ( T )e.NewItems[ i ];

          this.RaiseDataReplaced( oldItem, newItem );
        }
      }
      else if( e.Action == DataSourceChangedAction.ItemChanged )
      {
        foreach( ItemChangedDescription itemChanged in e.ItemsChanged )
        {
          ItemChangedInfo itemChangedInfo;

          if( itemChanged.ChangedProperties.Count == 0 )
          {
            itemChangedInfo = new ItemChangedInfo( itemChanged.Item );
          }
          else
          {
            List<string> propertiesName = new List<string>();
            Dictionary<string, object> oldValues = new Dictionary<string, object>();

            foreach( PropertyChangedDescription propertyChanged in itemChanged.ChangedProperties )
            {
              string propertyName = propertyChanged.PropertyName;
              propertiesName.Add( propertyName );

              object oldValue = propertyChanged.OldValue;
              if( oldValue != PropertyChangedDescription.UnsetValue )
              {
                oldValues.Add( propertyName, oldValue );
              }
            }

            itemChangedInfo = ( oldValues.Count > 0 )
                                ? new ItemChangedInfo( itemChanged.Item, propertiesName )
                                : new ItemChangedInfo( itemChanged.Item, propertiesName, oldValues );
          }

          this.RaiseDataItemChanged( itemChangedInfo );
        }
      }
      else if( e.Action == DataSourceChangedAction.Reset )
      {
        this.RaiseDataReset();
      }
      else
      {
        throw new DataInternalException();
      }
    }


    #region IPendingRequests Members

    public bool HasPendingRequests
    {
      get
      {
        return !this.IsQueueEmpty;
      }
    }

    public event EventHandler HasPendingRequestsChanged;

    private void RaiseHasPendingRequestsChanged()
    {
      if( this.HasPendingRequestsChanged != null )
      {
        this.HasPendingRequestsChanged( this, EventArgs.Empty );
      }
    }

    #endregion

    #region Private Fields

    private readonly IAsyncEnumerator<T> m_childEnumerator; //null
    private readonly INotifyDataSourceChanged m_dataSourceChanged;
    private List<ReactiveOperationInfo<T>> m_operationsQueue = new List<ReactiveOperationInfo<T>>();
    private readonly List<StatOperationInfo> m_statQueries = new List<StatOperationInfo>();
    private readonly DispatcherTimer m_processOperationTimer; //null
    private bool m_isProcessingOperation; //false
    private bool m_disposed; //false

    private readonly ThrottlingHandler m_throttlingHandler; //null

    #endregion

    #region StatOperationInfo

    private class StatOperationInfo
    {
      internal StatOperationInfo( StatParameters parameters, DataSourceContext context )
      {
        if( context == null )
          throw new ArgumentNullException( "context" );

        m_initialParameters = parameters;
        m_parameters = new StatParameters(
                         parameters.StatFunctions,
                         parameters.TargetPath,
                         context );
      }

      internal StatParameters InitialParameters
      {
        get
        {
          return m_initialParameters;
        }
      }

      internal StatParameters Parameters
      {
        get
        {
          return m_parameters;
        }
      }

      internal StatResults DummyResults
      {
        get;
        set;
      }

      internal StatResults RealResults
      {
        get;
        set;
      }

      private readonly StatParameters m_initialParameters;
      private readonly StatParameters m_parameters;
    }

    #endregion
  }
}
